package com.corpgovernment.common.mq.producer;

import com.alibaba.fastjson.JSONObject;
import com.corpgovernment.common.mq.MessageWrapper;
import com.corpgovernment.common.utils.TenantUtils;
import com.ctrip.corp.obt.generic.core.context.TenantContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * @author jhxue
 * @create 2021-12-10-10:26
 */
@Component
@Slf4j
public class CommonMQProducer {

	@Autowired
	private DefaultMQProducer defaultMQProducer;

	/**
	 * 推送消息
	 */
	public SendResult send(Message message) {
		try {
			String topic = message.getTopic();
			String newTopic = TenantUtils.getTenantTopic(topic);
			message.setTopic(newTopic);

			MessageWrapper messageWrapper = new MessageWrapper(message);
			messageWrapper.setTenantId(TenantContext.getTenantId());
			messageWrapper.setClientId(TenantContext.getClientId());
			log.info("开始推送消息：{}", JSONObject.toJSONString(message));
			SendResult send = defaultMQProducer.send(message);
			log.info("推送消息结果状态：{}", JSONObject.toJSONString(send));
			return send;
		} catch (Exception e) {
			log.error("推送消息结果失败", e);
		}
		return null;
	}

	/**
	 * 推送消息
	 */
	public SendResult send(Message message, MessageQueueSelector selector, Object arg) {
		try {
			String topic = message.getTopic();
			String newTopic = TenantUtils.getTenantTopic(topic);
			message.setTopic(newTopic);
			log.info("开始推送消息：{}", JSONObject.toJSONString(message));
			SendResult send = defaultMQProducer.send(message, selector, arg);
			log.info("推送消息结果状态：{}", JSONObject.toJSONString(send));
			return send;
		} catch (Exception e) {
			log.error("推送消息结果失败", e);
		}
		return null;
	}

	public void sendOneway(Message message) throws Exception {
		try {
			String topic = message.getTopic();
			String newTopic = TenantUtils.getTenantTopic(topic);
			message.setTopic(newTopic);
			log.info("开始推送消息：{}", JSONObject.toJSONString(message));
			defaultMQProducer.sendOneway(message);
			log.info("推送完成");
		} catch (Exception e) {
			log.error("推送消息结果失败", e);
		}
	}
}
